/******************************************
 * Author: Dave Spadea, Spadea Enterprises, Inc.
 * Date: November 2010
 * 
 * This code is released by Spadea Enterprises, Inc for general consumption 
 * under the Apache Software License.
 * 
 */


package net.spadea.framework.service;

import net.spadea.framework.messaging.*;

public abstract class FrameworkService extends ServiceBase {
	
	BrokerSubscriber subscriber = new BrokerSubscriber();
	
	public FrameworkService() {
		super();
		
		subscriber.connect(getBrokerUrl());
		Channel queue = listenChannel();
		log.info("Listening for events on " + queue.toString());
		subscriber.subscribe(listenChannel());
		
	}
	
	/***
	 * This method can be overridden to return a more appropriate value for your system.
	 * The default is to return the value of FRAMEWORK_BROKER_URL environment variable.
	 * @return
	 */
	public String getBrokerUrl() {
		return System.getenv("FRAMEWORK_BROKER_URL");
	}
	
	public void run() {

		while(true) {
			
			String yaml_text = subscriber.getMessage();
			
			Message msg = new GenericMessage();
			log.debug("Deserializing message yaml:\n" + yaml_text);
			msg.load(yaml_text);
			
			log.debug("Invoking message handler callback...");
			try {
				processMessage(msg);
			} catch (Exception e) {
				log.error("Message handler threw exception:\n" + e.getClass().getSimpleName() + ": " + e.getMessage());
				
				StringBuilder stacktrace = new StringBuilder();
				for (StackTraceElement ste : e.getStackTrace()) {
					stacktrace.append(ste.getFileName() + 
							":" + ste.getClassName() + 
							":" + ste.getMethodName() + "()" + 
							":" + ste.getLineNumber() + "\n");
				}
				
				log.error(stacktrace.toString());
				
				Integer errorCount;
				
				try {
					errorCount = (Integer) msg.get("ErrorInfo", "error_route_count");
				} catch (NumberFormatException nfe) {
					errorCount = null;
				}
				
				if (errorCount == null)
					errorCount = 1;
				else
					errorCount++;
				
				if (errorCount > 3) {
					log.error("Message looped 3 times as an error. Not routing.");
				} else {
					
					msg.set("ErrorInfo", "error_route_count", errorCount);
					msg.setHeader("is_error", "true");
					msg.set("ErrorInfo", "stacktrace", stacktrace.toString());
					msg.set("ErrorInfo", "message", e.getMessage());
					msg.setStatus(e.getClass().getCanonicalName()); 
					
					msg.send();
				}
			}
			
		}
		
	}

	public abstract void processMessage(Message msg) throws Exception;
	
	/** 
	 * Override listen_queue to provide a custom queue name. It must return an initialized QueueChannel or TopicChannel.
	 * @return
	 */
	public Channel listenChannel() {
		return new QueueChannel(this.getClass().getName());
	}

}
